/**
 * 
 */
package com.fang.www.join.two.three;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.fang.www.join.one.EmpMapper;
import com.fang.www.join.one.EmpReducer;
import com.fang.www.join.one.Emplyee;
import com.fang.www.join.one.EmplyeeJob;

/**
 * @author 王刚
 *
 */
public class MR {
	private static String outputUrl ="/wg/joinThree/output";
	private static String inputUrl = "/wg/joinThree/testExample";
	private static String hdfs = "hdfs://192.168.8.70:8020";
	public static void main(String[] arg) throws Exception {
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", hdfs);
		FileSystem fs =FileSystem.get(conf);
		Job job = Job.getInstance(conf, "join3");
		job.setJarByClass(MR.class);
		
		job.setMapOutputKeyClass(UserKey.class);
		job.setMapOutputValueClass(User.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);
		
		job.setMapperClass(Map.class);
		job.setReducerClass(Reduce.class);
		job.setGroupingComparatorClass(GroupComparator.class);
		fs.delete(new Path(outputUrl),true);
		FileInputFormat.addInputPath(job, new Path(inputUrl));
		FileOutputFormat.setOutputPath(job, new Path(outputUrl));
		job.waitForCompletion(true);
		
	}
}
